import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

/**
  * @author: yangShen
  * @Description: 布隆过滤器实现 UV(独立访客数) 统计       ---------针对大批量数据，例如几百G的数据
  * @Date: 2020/4/30 12:49 
  */
object UvWithBloom {
  //独立访客数超大批量时，要统计独立访客量，此时要对数据根据访客去重，存储放到内存时不行，要放到redis,直接用redis操作
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //用相对路径定义数据源
    val resource = getClass.getResource("/UserBehavior.csv")
    val dataStream = environment.readTextFile(resource.getPath)
      .map(data => {
        val dataArray = data.split(",")
        UserBehavior(dataArray(0).trim.toLong, dataArray(1).trim.toLong, dataArray(2).trim.toInt, dataArray(3).trim, dataArray(4).trim.toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)
      .filter(_.behavior == "pv")   //只统计pv操作
      .map( data => ("dummyKey", data.userId) )
      .keyBy(_._1)  //根据dummyKey进行分组
      .timeWindow(Time.hours(1))  //时间窗口，滚动窗口
      //触发器，一般默认窗口关闭时触发；若是到了窗口关闭时触发等于说把整个窗口的数据(大批量)都缓存到了内存(此时内存是放不下那个大的数据的)中，也就没有放到redis的必要了，
      //应该改为：来一次数据就触发一次
      .trigger(new MyTrigger())
      .process(new UvCountWithBloomProcessFunction())

    dataStream.print("UV count")

    environment.execute("uv with bloom job")
  }
}

//自定义窗口触发器            -----类型[二元组 是map之后的数据类型, ]
class MyTrigger() extends Trigger[ (String, Long), TimeWindow ]{
  override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  //因为没有注册定时器，所以不用去清除
  override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = {}

  override def onElement(element: (String, Long), timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = {
    //每来一条数据就，就直接触发窗口操作，并清空所有状态, 所以将count值存到redis
    TriggerResult.FIRE_AND_PURGE
  }
}

//定义一个布隆过滤器
//几十个G的数据，压缩下来只有几十 M
//1b(bit位)1个位能存储就是0或1：一个字节是8位，1G = 1024M * 1024KB * 1024B * 8bit = 8589934592
//Int 的 -2^31 (-2,147,483,648) 到 2^31 - 1 (2,147,483,647
class Bloom(size: Long) extends Serializable{
  //16M = 16 * 1024KB * 1024B * 8b = 2^27 ; 解释一：4个2 * 10个2 * 10个2 * 3个2 ; 解释二算法：  1024 = 2^10   16 = 2^4  8 = 2^3
  //位图的总大小：16M
  private val cap = if (size > 0) size else 1 << 27   //左移27位，1后面有27个0

  //定义hash函数，返回一个long类型的结果      ----参数(传递字符串值, 随机数：稍微差一点点最后的分布也会差很大 避免碰撞)
  def hash(value: String, seed: Int): Long = {
    var result = 0L
    //遍历value,对每一位进行算法，在result进行叠加
    for(i <- 0 until value.length){
      //对字符串的每一位进行处理，字符串的每一位都能得到一个特殊的值
      result += result * seed + value.charAt(i)
    }
    //cap : 1000000000000000000000000000, cap-1 : 1111111111111111111111111
    result & ( cap -1 )   //结果在 0 ~ 1111111111111111111111111(27个1) 之间
  }
}

//问题：UniqueVisitor.scala把数据都放到了内存中(费空间)，UvWithBloom.scala把数据一条一条的读取(费时间)并放到了redis
//优化：应该读取小批量的数据后，再放到redis中，结合的内存和redis

//                                        类型[输入参数为二元组 根据.map的返回决定, 输出UvCount, key为.keyBy(_._1), 定义的时间窗口.timeWindow]
class UvCountWithBloomProcessFunction extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow]{

  //定义redis连接, redis一共保存两个东西 一个是位图(布隆过滤器使用) 一个是count(redis表)
  lazy val jedis = new Jedis("localhost",6379)
  //左移29位，2^9 M = 512M, 512b / 8 = 64B
  lazy val bloom = new Bloom(1 << 29)    //个人理解：左移29位就是在左移27位多移了两位就 = 16M * 2个2 = 64M

  //命令验证查看redis：1:keys *; 2:hgetall count;
  override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {
    //位图的存储方式，以窗口的时间戳作为key,一个key对应一个位图，key是windowEnd,value是bitmap
    val storeKey = context.window.getEnd.toString
    var count = 0L
    //把每个窗口的uv count值也存入名为count的redis表, 存放内容为( windowEnd -> uvCount )，所以要先从redis中读取
    if (jedis.hget("count",storeKey) != null){
      count = jedis.hget("count",storeKey).toLong
    }
    //用布隆过滤器判断当前用户是否存在,MyTrigger中设置的是一条一条的处理，所以elements中只有一条,布隆过滤器中为String，所以转换.toString
    val userId = elements.last._2.toString
    //对userId进行hash计算，根据hash值去位图中找到对应的count, 61为随机值
    val offset = bloom.hash(userId, 61)
    //定义一个标识位，判断redis位图中有没有这一位, .getbit获取存储某一个值的某一位
    val isExist = jedis.getbit(storeKey, offset)
    if (! isExist){
      //如果不存在，位图对应位置置位1，count进行 count + 1
      jedis.setbit(storeKey,offset,true)
      jedis.hset("count", storeKey, (count +1).toString)
      out.collect(UvCount(storeKey.toLong, count + 1))
    }else {
      out.collect(UvCount(storeKey.toLong, count))
    }
  }
}
